package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.atguigu.gmall.realtime.utils.MySqlUtil;
import com.sun.xml.internal.bind.v2.TODO;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: Dwd02_TradeCartAdd
 * Package: com.atguigu.gmall.realtime.app.dwd.db
 * Description:      交易域加购事务事实表
 *          提取加购操作生成加购表，并将字典表中的相关维度退化到加购表中，写出到 Kafka 对应主题。
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/4/29 17:45
 * @Version 1.0
 */

//数据流：web/app -> Mysql -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程 序：Mock -> Mysql -> Maxwell -> Kafka(ZK) -> Dwd02_TradeCartAdd -> Kafka(ZK)
public class Dwd02_TradeCartAdd {
    public static void main(String[] args) {

        //TODO 1.获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // TODO 配置环境信息生产环境得有

//         env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");


        //TODO 2.使用flinksql方式从kafka读取 topic_db数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_cart_add_221109"));

        //TODO 3.过滤出加购数据
        Table cartInfoTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['sku_id'] sku_id,\n" +
                "    `data`['cart_price'] cart_price,\n" +
                "    `data`['sku_num'] sku_num,\n" +
                "    `data`['sku_name'] sku_name,\n" +
                "    `data`['is_checked'] is_checked,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['operate_time'] operate_time,\n" +
                "    `data`['is_ordered'] is_ordered,\n" +
                "    `data`['order_time'] order_time,\n" +
                "    `data`['source_type'] source_type,\n" +
                "    `data`['source_id'] source_id,\n" +
                "    `pt`\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='cart_info'\n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("cart_info", cartInfoTable);

        //TODO 4.从mysql加载base_dic作为lookup表
        tableEnv.executeSql(MySqlUtil.getMysqlBaseDicDDL());

        //TODO 5.关联两表
        Table resultTable = tableEnv.sqlQuery("" +
                        "select\n" +
                        "    cart_info.id,\n" +
                        "    cart_info.user_id,\n" +
                        "    cart_info.sku_id,\n" +
                        "    cart_info.cart_price,\n" +
                        "    cart_info.sku_num,\n" +
                        "    cart_info.sku_name,\n" +
                        "    cart_info.is_checked,\n" +
                        "    cart_info.create_time,\n" +
                        "    cart_info.operate_time,\n" +
                        "    cart_info.is_ordered,\n" +
                        "    cart_info.order_time,\n" +
                        "    cart_info.source_type,\n" +
//                "    dic.dic_name,\n" +
                        "    cart_info.source_id\n" +
                        "from cart_info\n"
//                        +
//                "join base_dic FOR SYSTEM_TIME AS OF cart_info.pt AS dic\n" +
//                "on cart_info.`source_type`=dic.`dic_code`"
        );
        tableEnv.createTemporaryView("result_table", resultTable);

        //TODO 6.创建kafka dwd层加购事务主题
        tableEnv.executeSql("" +
                "create table dwd_cart_info(\n" +
                "    `id` string,\n" +
                "    `user_id` string,\n" +
                "    `sku_id` string,\n" +
                "    `cart_price` string,\n" +
                "    `sku_num` string,\n" +
                "    `sku_name` string,\n" +
                "    `is_checked` string,\n" +
                "    `create_time` string,\n" +
                "    `operate_time` string,\n" +
                "    `is_ordered` string,\n" +
                "    `order_time` string,\n" +
                "    `source_type` string,\n" +
//                "    `dic_name` string,\n" +
                "    `source_id` string\n" +
                ")" + MyKafkaUtil.getKafkaSinkConnOption("dwd_trade_cart_add"));

        //TODO 7.插入数据
        tableEnv.executeSql("insert into dwd_cart_info select * from result_table");
    }
}
